package sunyu.toolkit.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.slf4j.Logger;
import sunyu.toolkit.core.ExceptionKit;
import sunyu.toolkit.core.LogKit;

import java.io.IOException;
import java.net.URI;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author 孙宇
 */
public class HdfsKit {

    private static final Logger logger = LogKit.getLogger();
    private static final Map<String, FileSystem> fileSystemPool = new ConcurrentHashMap<>();
    private static Configuration conf;
    private static String HADOOP_USER_NAME;
    private static String defaultFS;

    /**
     * 初始化
     *
     * @param HADOOP_USER_NAME 用户名
     * @param defaultFS        hdfs地址
     */
    public static void init(String HADOOP_USER_NAME, String defaultFS) {
        //System.setProperty("HADOOP_USER_NAME", HADOOP_USER_NAME);
        HdfsKit.HADOOP_USER_NAME = HADOOP_USER_NAME;
        HdfsKit.defaultFS = defaultFS;
        conf = new Configuration();
        conf.set("fs.defaultFS", defaultFS);
        conf.setBoolean("dfs.support.append", true);
    }

    public static void destroy() {
        for (Map.Entry<String, FileSystem> entry : fileSystemPool.entrySet()) {
            try {
                entry.getValue().close();
            } catch (IOException e) {
                logger.error(ExceptionKit.getStackTrace(e));
            }
        }
    }

    public static FileSystem getFileSystem(String path) {
        if (fileSystemPool.get(path) == null) {
            String url;
            if (path.indexOf(defaultFS) == 0) {
                url = path;
            } else {
                url = defaultFS + path;
            }
            URI uri = URI.create(url);
            try {
                FileSystem fs = FileSystem.get(uri, conf, HADOOP_USER_NAME);
                fileSystemPool.put(path, fs);
            } catch (IOException e) {
                logger.error(ExceptionKit.getStackTrace(e));
            } catch (InterruptedException e) {
                logger.error(ExceptionKit.getStackTrace(e));
            }
        }
        return fileSystemPool.get(path);
    }

    /**
     * 获取HDFS集群上所有节点名称信息
     *
     * @return 节点信息集合
     */
    public static DatanodeInfo[] getDataNodeStats() {
        DistributedFileSystem hdfs = (DistributedFileSystem) getFileSystem("/");
        DatanodeInfo[] dataNodeStats = new DatanodeInfo[0];
        try {
            dataNodeStats = hdfs.getDataNodeStats();
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return dataNodeStats;
    }


    /**
     * 获得文件状态列表
     *
     * @param path 路径
     * @return 文件状态列表
     */
    public static FileStatus[] listStatus(String path) {
        FileSystem fs = getFileSystem(path);
        FileStatus fileStatus[] = new FileStatus[0];
        try {
            fileStatus = fs.listStatus(new Path(path));
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return fileStatus;
    }

    /**
     * 获得文件状态
     *
     * @param path 路径
     * @return 文件状态
     */
    public static FileStatus getFileStatus(String path) {
        FileSystem fs = getFileSystem(path);
        try {
            return fs.getFileStatus(new Path(path));
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return null;
    }

    /**
     * 查找某个文件在HDFS集群的位置
     *
     * @param path 路径
     * @return BlockLocation集合
     */
    public static BlockLocation[] getFileBlockLocations(String path) {
        BlockLocation[] blockLocation = new BlockLocation[0];
        FileSystem fs = getFileSystem(path);
        FileStatus fileStatus = getFileStatus(path);
        try {
            blockLocation = fs.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return blockLocation;
    }

    /**
     * 上传本地文件到hdfs
     *
     * @param localFilePath 本地路径
     * @param dst           目标路径
     * @return 是否上传成功
     */
    public static boolean copyFromLocalFile(String localFilePath, String dst) {
        boolean b = false;
        FileSystem fs = getFileSystem(dst);
        Path src = new Path(localFilePath);
        try {
            fs.copyFromLocalFile(src, new Path(dst));
            b = true;
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return b;
    }

    /**
     * 创建文件
     *
     * @param path 路径
     * @return 是否创建成功
     */
    public static boolean create(String path) {
        boolean b = false;
        FileSystem fs = getFileSystem(path);
        Path dfs = new Path(path);
        try (FSDataOutputStream outputStream = fs.create(dfs);) {
            outputStream.write(new byte[0]);
            b = true;
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return b;
    }

    /**
     * 创建文件夹
     *
     * @param path 路径
     * @return 是否创建成功
     */
    public static boolean mkdirs(String path) {
        boolean b = false;
        FileSystem fs = getFileSystem(path);
        try {
            fs.mkdirs(new Path(path));
            b = true;
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return b;
    }

    /**
     * 重命名文件
     *
     * @param path    原路径
     * @param newPath 新路径
     * @return 是否重命名成功
     */
    public static boolean rename(String path, String newPath) {
        FileSystem fs = getFileSystem(path);
        try {
            return fs.rename(new Path(path), new Path(newPath));
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return false;
    }

    /**
     * 删除文件或目录
     *
     * @param path      路径
     * @param recursive 是否递归删除
     * @return 是否删除成功
     */
    public static boolean delete(String path, boolean recursive) {
        FileSystem fs = getFileSystem(path);
        try {
            return fs.delete(new Path(path), recursive);
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return false;
    }

    /**
     * 判断是否存在
     *
     * @param path 路径
     * @return 是否存在
     */
    public static boolean exists(String path) {
        FileSystem fs = getFileSystem(path);
        try {
            return fs.exists(new Path(path));
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return false;
    }

    /**
     * 读取文件
     *
     * @param path 路径
     * @return 文件流
     */
    public static FSDataInputStream read(String path) {
        FileSystem fs = getFileSystem(path);
        try {
            return fs.open(new Path(path));
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return null;
    }

    /**
     * 追加内容
     *
     * @param path    路径
     * @param content 内容
     * @return 是否追加成功
     */
    public static boolean appendChars(String path, String content) {
        boolean b = false;
        FileSystem fs = getFileSystem(path);
        try (FSDataOutputStream os = fs.append(new Path(path));) {
            os.writeChars(content);
            os.flush();
            b = true;
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return b;
    }

}
